/*
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later
 * version.
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
 * details.
 * 
 * You should have received a copy of the GNU General Public License along with
 * this program. If not, see <http://www.gnu.org/licenses/>.
 */
package com.aionemu.commons.network.mmocore;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SocketChannel;

import javolution.text.TextBuilder;

import org.apache.log4j.Logger;

import com.aionemu.commons.network.mmocore.FloodManager.ErrorMode;
import com.aionemu.commons.network.mmocore.FloodManager.Result;
import com.aionemu.commons.network.mmocore.handler.AbstractPacketHandlerFactory;

/**
 * @author KenM<BR>
 *         Parts of design based on networkcore from WoodenGil
 */
public abstract class SelectorThread<T extends MMOConnection<T>>
{
	public static final Logger			log	= Logger.getLogger(SelectorThread.class);

	private final AcceptorThread<T>		_acceptorThread;
	private final ReadWriteThread<T>[]	_readWriteThreads;

	@SuppressWarnings("unchecked")
	protected SelectorThread(SelectorConfig sc, AbstractPacketHandlerFactory<T> packetHandler) throws IOException
	{
		_acceptorThread = new AcceptorThread<T>("AcceptorThread", this, sc);
		_readWriteThreads = new ReadWriteThread[sc.getSelectorThreadCount()];

		for (int i = 0; i < _readWriteThreads.length; i++)
			_readWriteThreads[i] = new ReadWriteThread<T>("ReadWriteThread-" + (i + 1), this, sc, packetHandler);
	}

	public final void openServerSocket(InetSocketAddress address) throws IOException
	{
		getAcceptorThread().openServerSocket(address);
	}

	private AcceptorThread<T> getAcceptorThread()
	{
		return _acceptorThread;
	}

	private ReadWriteThread<T>[] getReadWriteThreads()
	{
		return _readWriteThreads;
	}

	private int	_readWriteThreadIndex;

	final ReadWriteThread<T> getReadWriteThread()
	{
		return getReadWriteThreads()[_readWriteThreadIndex++ % getReadWriteThreads().length];
	}

	public final void start()
	{
		getAcceptorThread().start();

		for (ReadWriteThread<T> readWriteThread : getReadWriteThreads())
			readWriteThread.start();
	}

	public final void shutdown() throws InterruptedException
	{
		getAcceptorThread().shutdown();

		for (ReadWriteThread<T> readWriteThread : getReadWriteThreads())
			readWriteThread.shutdown();
	}

	// ==============================================

	protected abstract T createClient(SocketChannel socketChannel) throws ClosedChannelException;

	protected abstract void executePacket(ReceivablePacket<T> packet);

	// ==============================================

	private final FloodManager	_accepts;
	private final FloodManager	_packets;
	private final FloodManager	_errors;

	{
		// TODO fine tune
		_accepts = new FloodManager(1000, // 1000 msec per tick
				new FloodManager.FloodFilter(10, 20, 10), // short period
				new FloodManager.FloodFilter(30, 60, 60)); // long period

		_packets = new FloodManager(1000, // 1000 msec per tick
				new FloodManager.FloodFilter(250, 300, 2));

		_errors = new FloodManager(200, // 200 msec per tick
				new FloodManager.FloodFilter(10, 10, 1));
	}

	protected String getVersionInfo()
	{
		return "";
	}

	protected boolean acceptConnectionFrom(SocketChannel sc)
	{
		final String host = sc.socket().getInetAddress().getHostAddress();

		final Result isFlooding = _accepts.isFlooding(host, true);

		switch (isFlooding)
		{
			case REJECTED:
			{
				// TODO punish, warn, log, etc
				log.warn("Rejected connection from " + host);
				return false;
			}
			case WARNED:
			{
				// TODO punish, warn, log, etc
				log.warn("Connection over warn limit from " + host);
				return true;
			}
			default:
				return true;
		}
	}

	public void report(ErrorMode mode, T client, ReceivablePacket<T> packet, Throwable throwable)
	{
		final Result isFlooding = _errors.isFlooding(client.getValidUID(), true);

		final TextBuilder sb = TextBuilder.newInstance();
		if (isFlooding != Result.ACCEPTED)
		{
			sb.append("Flooding with ");
		}
		sb.append(mode);
		sb.append(": ");
		sb.append(client);
		final String versionInfo = getVersionInfo();
		if (versionInfo != null && !versionInfo.isEmpty())
		{
			sb.append(" - ");
			sb.append(versionInfo);
		}

		if (throwable != null)
			log.info(sb.toString(), throwable);
		else
			log.info(sb.toString());

		TextBuilder.recycle(sb);
		//if (isFlooding != Result.ACCEPTED)
		//{
		//	// TODO punish, warn, log, etc
		//}
	}

	protected boolean canReceivePacketFrom(T client, int opcode)
	{
		final String key = client.getValidUID();

		switch (Result.max(_packets.isFlooding(key, true), _errors.isFlooding(key, false)))
		{
			case REJECTED:
			{
				// TODO punish, warn, log, etc
				log.warn("Rejected packet (0x" + Integer.toHexString(opcode) + ") from " + client);
				return false;
			}
			case WARNED:
			{
				// TODO punish, warn, log, etc
				log.warn("Packet over warn limit (0x" + Integer.toHexString(opcode) + ") from " + client);
				return true;
			}
			default:
				return true;
		}
	}
}
